package com.persagy.iot.func

import com.persagy.iot.bean.IOTData
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * 数据三种状态
 * 1、正常数据，数据产生时间与实际入库的时间差小于 1800s
 * 2、迟到数据，数据产生时间与实际入库的时间差大于 1800s，小于3600s
 * 3、丢数
 *    a、数据未到：state的时间触发器超过 3600s，侧输出流输出 rowKey 报警信息
 *    b、数据已到：数据产生时间与实际入库的时间差大于 3600s
 *    c、数据的产生时间如果是晚于进入系统时间，那么根据业务规则的阈值 3600s(暂定)，超过阈值的数据按异常数据处理
 */
class OriginalDataAlarm extends KeyedProcessFunction[String, IOTData, IOTData] {

  /** 最新数据状态 */
  lazy val alarmState: ValueState[IOTData] = getRuntimeContext.getState(new ValueStateDescriptor[IOTData]("alarm-IOTData", classOf[IOTData]))

  /** 丢数时间触发器 */
  lazy val loseTime: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("loseTime", classOf[Long]))

  override def processElement(iotData: IOTData,
                              ctx: KeyedProcessFunction[String, IOTData, IOTData]#Context,
                              out: Collector[IOTData]): Unit = {

    val now: Long = System.currentTimeMillis() + (10 * 1000L)

    // 数据第一次过来，定义触发器
    if (alarmState.value() != null && loseTime.value() != 0L) {
      ctx.timerService().deleteProcessingTimeTimer(loseTime.value())
    }

    loseTime.update(now)
    alarmState.update(iotData)
    ctx.timerService().registerProcessingTimeTimer(loseTime.value())

    out.collect(iotData)
  }

  override def onTimer(timestamp: Long,
                       ctx: KeyedProcessFunction[String, IOTData, IOTData]#OnTimerContext,
                       out: Collector[IOTData]): Unit = {

    // 将丢数的点位放入侧输出流
    val lateDataTag = new OutputTag[IOTData]("late-data")
    ctx.output(lateDataTag, alarmState.value())

    alarmState.clear()
    loseTime.clear()
  }
}
